// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/dwio/common/input_stream.h>

#include <melon/container/f14_map.h>
#include <turbo/flags/flag.h>
#include <turbo/log/logging.h>
#include <cstdint>
#include <functional>
#include <istream>
#include <numeric>
#include <stdexcept>
#include <string_view>
#include <type_traits>

#include <pollux/common/time/timer.h>
#include <pollux/common/exception/exception.h>

using ::kumo::pollux::common::Region;

namespace kumo::pollux::dwio::common {
    namespace {
        int64_t totalBufferSize(const std::vector<melon::Range<char *> > &buffers) {
            int64_t bufferSize = 0;
            for (auto &buffer: buffers) {
                bufferSize += buffer.size();
            }
            return bufferSize;
        }
    } // namespace

    melon::SemiFuture<uint64_t> InputStream::readAsync(
        const std::vector<melon::Range<char *> > &buffers,
        uint64_t offset,
        LogType logType) {
        try {
            read(buffers, offset, logType);
            uint64_t size = 0;
            for (auto &range: buffers) {
                size += range.size();
            }
            return melon::SemiFuture<uint64_t>(size);
        } catch (const std::exception &e) {
            return melon::makeSemiFuture<uint64_t>(e);
        }
    }

    ReadFileInputStream::ReadFileInputStream(
        std::shared_ptr<pollux::ReadFile> readFile,
        const MetricsLogPtr &metricsLog,
        IoStatistics *stats,
        filesystems::File::IoStats *fsStats)
        : InputStream(readFile->getName(), metricsLog, stats, fsStats),
          readFile_(std::move(readFile)) {
    }

    void ReadFileInputStream::read(
        void *buf,
        uint64_t length,
        uint64_t offset,
        MetricsLog::MetricsType purpose) {
        POLLUX_CHECK_NOT_NULL(buf);
        logRead(offset, length, purpose);
        uint64_t readTimeUs{0};
        std::string_view readData; {
            MicrosecondTimer timer(&readTimeUs);
            readData = readFile_->pread(offset, length, buf, fsStats_);
        }
        if (stats_) {
            stats_->incRawBytesRead(length);
            stats_->incTotalScanTime(readTimeUs * 1'000);
        }

        POLLUX_CHECK_EQ(
            readData.size(),
            length,
            "Should read exactly as requested. File name: {}, offset: {}, length: {}, read: {}",
            getName(),
            offset,
            length,
            readData.size());
    }

    void ReadFileInputStream::read(
        const std::vector<melon::Range<char *> > &buffers,
        uint64_t offset,
        LogType logType) {
        const int64_t bufferSize = totalBufferSize(buffers);
        logRead(offset, bufferSize, logType);
        const auto size = readFile_->preadv(offset, buffers, fsStats_);
        POLLUX_CHECK_EQ(
            size,
            bufferSize,
            "Should read exactly as requested. File name: {}, offset: {}, length: {}, read: {}",
            getName(),
            offset,
            bufferSize,
            size);
    }

    melon::SemiFuture<uint64_t> ReadFileInputStream::readAsync(
        const std::vector<melon::Range<char *> > &buffers,
        uint64_t offset,
        LogType logType) {
        const int64_t bufferSize = totalBufferSize(buffers);
        logRead(offset, bufferSize, logType);
        return readFile_->preadvAsync(offset, buffers, fsStats_);
    }

    bool ReadFileInputStream::hasReadAsync() const {
        return readFile_->hasPreadvAsync();
    }

    void ReadFileInputStream::vread(
        melon::Range<const pollux::common::Region *> regions,
        melon::Range<melon::IOBuf *> iobufs,
        const LogType purpose) {
        POLLUX_CHECK_GT(regions.size(), 0, "regions to read can't be empty");
        const size_t length = std::accumulate(
            regions.cbegin(),
            regions.cend(),
            size_t(0),
            [&](size_t acc, const auto &r) { return acc + r.length; });
        logRead(regions[0].offset, length, purpose);
        auto readStartMicros = getCurrentTimeMicro();
        readFile_->preadv(regions, iobufs, fsStats_);
        if (stats_) {
            stats_->incRawBytesRead(length);
            stats_->incTotalScanTime((getCurrentTimeMicro() - readStartMicros) * 1000);
        }
    }

    const std::string &InputStream::getName() const {
        return path_;
    }

    void InputStream::logRead(uint64_t offset, uint64_t length, LogType purpose) {
        metricsLog_->logRead(
            0, "readFully", getLength(), 0, 0, offset, length, purpose, 1, 0);
    }
} // namespace kumo::pollux::dwio::common
